package com.friendsystem.dao;

import com.friendsystem.model.User;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class UserDao {
    private Connection connection;
    private Table userTable;

    public UserDao() throws IOException {
        Configuration config = HBaseConfiguration.create();
        config.addResource("hadoop-site.xml");
        connection = ConnectionFactory.createConnection(config);
        userTable = connection.getTable(TableName.valueOf("users"));
    }

    public User getUserById(String userId) throws IOException {
        Get get = new Get(Bytes.toBytes(userId));
        Result result = userTable.get(get);

        if (result.isEmpty()) {
            return null;
        }

        User user = new User();
        user.setUserId(userId);
        user.setUsername(Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("username"))));
        user.setFriendCount(Bytes.toInt(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("friendCount"))));

        return user;
    }

    public List<User> getRandomUsers(int count) throws IOException {
        Scan scan = new Scan();
        ResultScanner scanner = userTable.getScanner(scan);

        List<User> users = new ArrayList<>();
        int i = 0;
        for (Result result : scanner) {
            if (i >= count) break;

            User user = new User();
            user.setUserId(Bytes.toString(result.getRow()));
            user.setUsername(Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("username"))));
            users.add(user);
            i++;
        }

        return users;
    }

    public void close() throws IOException {
        if (userTable != null) userTable.close();
        if (connection != null) connection.close();
    }
}